package org.lib.kafkahello.try1;

import java.util.Date;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

/**
 * 试验1 的调试接口
 * @author ben
 * @date 2021-08-24 18:57:58 CST
 */
@RestController
@RequestMapping("/try1")
@Slf4j
public class Try1Controller {

	@Autowired
	private KafkaTemplate<String, Object> kafkaTemplate;
	
	/**
	 * 调用接口发送消息
	 * @author ben
	 * @date 2021-08-24 16:31:33 CST
	 * @return
	 */
	@GetMapping(value="/send")
	public Boolean sendMsg() {
		IntStream.rangeClosed(0, 9).forEach(i->{
			log.info("sendMsg-{}", i);
			String msg = String.format("Try1Controller send msg-%d @%s", i, new Date());
			
			// 无key
//			kafkaTemplate.send(Try1Config.TOPIC_01, msg);

			ListenableFuture<SendResult<String, Object>> sendRetFuture = kafkaTemplate.send(Try1Config.TOPIC_01, msg);
			
			// 1、发送同步确认
			// 发生异常时，不终止，继续执行下一个发送
			// 发送消息确认：发送失败时怎么处理？停止继续发送、延迟后再发送？
//			boolean sendErr = false;
//			try {
//				SendResult<String, Object> sendRet  = sendRetFuture.get();
//				RecordMetadata rmd = sendRet.getRecordMetadata();
//				if (rmd != null) {
//					log.info("发送成功：rmd-topic={},partition={},offset={},ts={}", 
//							rmd.topic(), rmd.partition(), rmd.offset(), rmd.timestamp());
//				} else {
//					log.error("发送失败：topic={}, msg={}", Try1Config.TOPIC_01, msg);
//				}
//			} catch (InterruptedException | ExecutionException e) {
//				log.error("发生异常：发送失败，e={}", e);
//				sendErr = true;
//			}
//			if (sendErr) {
//				// 发生异常，休眠30秒
//				// 发送时，断掉服务器的网络，或停掉kafka，这里的30秒就有操作空间了
//				try {
//					TimeUnit.SECONDS.sleep(30);
//				} catch (InterruptedException e) {
//					log.error("发生异常：sleep-30secs, e={}", e);
//				}
//			}
			
			// 2、发送异步确认
			sendRetFuture.addCallback(new ListenableFutureCallback<SendResult<String,Object>>() {

				@Override
				public void onSuccess(SendResult<String,Object> result) {
					RecordMetadata rmd = result.getRecordMetadata();
					log.info("发送成功-回调：rmd-topic={},partition={},offset={},ts={}", 
							rmd.topic(), rmd.partition(), rmd.offset(), rmd.timestamp());
				}

				@Override
				public void onFailure(Throwable ex) {
					log.info("发送失败-回调：ex={}", ex);
				}
			});
			
			// 有key
//			kafkaTemplate.send(Try1Config.TOPIC_01, Try1Config.TOPIC_01_KEY, "key-" + msg);
			
			// partition=0
//			kafkaTemplate.send(Try1Config.TOPIC_01, 0, Try1Config.TOPIC_01_KEY, "key-partition-0-" + msg);
			
			// partition=1 超过主题的分区数量
			// Error: 程序卡住，最后超时终止
			// org.apache.kafka.common.errors.TimeoutException: Topic topic01 not present in metadata after 60000 ms.
//			kafkaTemplate.send(Try1Config.TOPIC_01, 1, Try1Config.TOPIC_01_KEY, "key-partition-1-" + msg);
			
			try {
				TimeUnit.SECONDS.sleep(2);
			} catch (InterruptedException e) {
				// 测试手动确认消息发送是否成功 时，每发送1条消息暂停2秒
				e.printStackTrace();
			}
		});
		
		return true;
	}
	
}
